home *** CD-ROM | disk | FTP | other *** search
/ Clickx 96 / Clickx 96.iso / software / tools / tool / xbmc-10.1.exe / addons / script.module.pysqlite / lib / pysqlite2 / test / transactions.py < prev    next >
Encoding:
Python Source  |  2009-10-19  |  7.2 KB  |  206 lines
  1. #-*- coding: ISO-8859-1 -*-
  2. # pysqlite2/test/transactions.py: tests transactions
  3. #
  4. # Copyright (C) 2005-2009 Gerhard HΣring <gh@ghaering.de>
  5. #
  6. # This file is part of pysqlite.
  7. #
  8. # This software is provided 'as-is', without any express or implied
  9. # warranty.  In no event will the authors be held liable for any damages
  10. # arising from the use of this software.
  11. #
  12. # Permission is granted to anyone to use this software for any purpose,
  13. # including commercial applications, and to alter it and redistribute it
  14. # freely, subject to the following restrictions:
  15. #
  16. # 1. The origin of this software must not be misrepresented; you must not
  17. #    claim that you wrote the original software. If you use this software
  18. #    in a product, an acknowledgment in the product documentation would be
  19. #    appreciated but is not required.
  20. # 2. Altered source versions must be plainly marked as such, and must not be
  21. #    misrepresented as being the original software.
  22. # 3. This notice may not be removed or altered from any source distribution.
  23.  
  24. import sys
  25. import os, unittest
  26. import pysqlite2.dbapi2 as sqlite
  27.  
  28. def get_db_path():
  29.     return "sqlite_testdb"
  30.  
  31. class TransactionTests(unittest.TestCase):
  32.     def setUp(self):
  33.         try:
  34.             os.remove(get_db_path())
  35.         except OSError:
  36.             pass
  37.  
  38.         self.con1 = sqlite.connect(get_db_path(), timeout=0.1)
  39.         self.cur1 = self.con1.cursor()
  40.  
  41.         self.con2 = sqlite.connect(get_db_path(), timeout=0.1)
  42.         self.cur2 = self.con2.cursor()
  43.  
  44.     def tearDown(self):
  45.         self.cur1.close()
  46.         self.con1.close()
  47.  
  48.         self.cur2.close()
  49.         self.con2.close()
  50.  
  51.         try:
  52.             os.unlink(get_db_path())
  53.         except OSError:
  54.             pass
  55.  
  56.     def CheckDMLdoesAutoCommitBefore(self):
  57.         self.cur1.execute("create table test(i)")
  58.         self.cur1.execute("insert into test(i) values (5)")
  59.         self.cur1.execute("create table test2(j)")
  60.         self.cur2.execute("select i from test")
  61.         res = self.cur2.fetchall()
  62.         self.failUnlessEqual(len(res), 1)
  63.  
  64.     def CheckInsertStartsTransaction(self):
  65.         self.cur1.execute("create table test(i)")
  66.         self.cur1.execute("insert into test(i) values (5)")
  67.         self.cur2.execute("select i from test")
  68.         res = self.cur2.fetchall()
  69.         self.failUnlessEqual(len(res), 0)
  70.  
  71.     def CheckUpdateStartsTransaction(self):
  72.         self.cur1.execute("create table test(i)")
  73.         self.cur1.execute("insert into test(i) values (5)")
  74.         self.con1.commit()
  75.         self.cur1.execute("update test set i=6")
  76.         self.cur2.execute("select i from test")
  77.         res = self.cur2.fetchone()[0]
  78.         self.failUnlessEqual(res, 5)
  79.  
  80.     def CheckDeleteStartsTransaction(self):
  81.         self.cur1.execute("create table test(i)")
  82.         self.cur1.execute("insert into test(i) values (5)")
  83.         self.con1.commit()
  84.         self.cur1.execute("delete from test")
  85.         self.cur2.execute("select i from test")
  86.         res = self.cur2.fetchall()
  87.         self.failUnlessEqual(len(res), 1)
  88.  
  89.     def CheckReplaceStartsTransaction(self):
  90.         self.cur1.execute("create table test(i)")
  91.         self.cur1.execute("insert into test(i) values (5)")
  92.         self.con1.commit()
  93.         self.cur1.execute("replace into test(i) values (6)")
  94.         self.cur2.execute("select i from test")
  95.         res = self.cur2.fetchall()
  96.         self.failUnlessEqual(len(res), 1)
  97.         self.failUnlessEqual(res[0][0], 5)
  98.  
  99.     def CheckToggleAutoCommit(self):
  100.         self.cur1.execute("create table test(i)")
  101.         self.cur1.execute("insert into test(i) values (5)")
  102.         self.con1.isolation_level = None
  103.         self.failUnlessEqual(self.con1.isolation_level, None)
  104.         self.cur2.execute("select i from test")
  105.         res = self.cur2.fetchall()
  106.         self.failUnlessEqual(len(res), 1)
  107.  
  108.         self.con1.isolation_level = "DEFERRED"
  109.         self.failUnlessEqual(self.con1.isolation_level , "DEFERRED")
  110.         self.cur1.execute("insert into test(i) values (5)")
  111.         self.cur2.execute("select i from test")
  112.         res = self.cur2.fetchall()
  113.         self.failUnlessEqual(len(res), 1)
  114.  
  115.     def CheckRaiseTimeout(self):
  116.         if sqlite.sqlite_version_info < (3, 2, 2):
  117.             # This will fail (hang) on earlier versions of sqlite.
  118.             # Determine exact version it was fixed. 3.2.1 hangs.
  119.             return
  120.         self.cur1.execute("create table test(i)")
  121.         self.cur1.execute("insert into test(i) values (5)")
  122.         try:
  123.             self.cur2.execute("insert into test(i) values (5)")
  124.             self.fail("should have raised an OperationalError")
  125.         except sqlite.OperationalError:
  126.             pass
  127.         except:
  128.             self.fail("should have raised an OperationalError")
  129.  
  130.     def CheckLocking(self):
  131.         """
  132.         This tests the improved concurrency with pysqlite 2.3.4. You needed
  133.         to roll back con2 before you could commit con1.
  134.         """
  135.         if sqlite.sqlite_version_info < (3, 2, 2):
  136.             # This will fail (hang) on earlier versions of sqlite.
  137.             # Determine exact version it was fixed. 3.2.1 hangs.
  138.             return
  139.         self.cur1.execute("create table test(i)")
  140.         self.cur1.execute("insert into test(i) values (5)")
  141.         try:
  142.             self.cur2.execute("insert into test(i) values (5)")
  143.             self.fail("should have raised an OperationalError")
  144.         except sqlite.OperationalError:
  145.             pass
  146.         except:
  147.             self.fail("should have raised an OperationalError")
  148.         # NO self.con2.rollback() HERE!!!
  149.         self.con1.commit()
  150.  
  151.     def CheckRollbackCursorConsistency(self):
  152.         """
  153.         Checks if cursors on the connection are set into a "reset" state
  154.         when a rollback is done on the connection.
  155.         """
  156.         con = sqlite.connect(":memory:")
  157.         cur = con.cursor()
  158.         cur.execute("create table test(x)")
  159.         cur.execute("insert into test(x) values (5)")
  160.         cur.execute("select 1 union select 2 union select 3")
  161.  
  162.         con.rollback()
  163.         try:
  164.             cur.fetchall()
  165.             self.fail("InterfaceError should have been raised")
  166.         except sqlite.InterfaceError, e:
  167.             pass
  168.         except:
  169.             self.fail("InterfaceError should have been raised")
  170.  
  171. class SpecialCommandTests(unittest.TestCase):
  172.     def setUp(self):
  173.         self.con = sqlite.connect(":memory:")
  174.         self.cur = self.con.cursor()
  175.  
  176.     def CheckVacuum(self):
  177.         self.cur.execute("create table test(i)")
  178.         self.cur.execute("insert into test(i) values (5)")
  179.         self.cur.execute("vacuum")
  180.  
  181.     def CheckDropTable(self):
  182.         self.cur.execute("create table test(i)")
  183.         self.cur.execute("insert into test(i) values (5)")
  184.         self.cur.execute("drop table test")
  185.  
  186.     def CheckPragma(self):
  187.         self.cur.execute("create table test(i)")
  188.         self.cur.execute("insert into test(i) values (5)")
  189.         self.cur.execute("pragma count_changes=1")
  190.  
  191.     def tearDown(self):
  192.         self.cur.close()
  193.         self.con.close()
  194.  
  195. def suite():
  196.     default_suite = unittest.makeSuite(TransactionTests, "Check")
  197.     special_command_suite = unittest.makeSuite(SpecialCommandTests, "Check")
  198.     return unittest.TestSuite((default_suite, special_command_suite))
  199.  
  200. def test():
  201.     runner = unittest.TextTestRunner()
  202.     runner.run(suite())
  203.  
  204. if __name__ == "__main__":
  205.     test()
  206.